package Spark2Core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2018/6/21.
  */
object Spark2RDD {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("ScalaWordCount").setMaster("local[4]")

    val sc = new SparkContext(conf)

    val rdd4 = sc.parallelize(Array("a b c", "d e f", "h i j"))

    val value: RDD[Array[String]] = rdd4.map(_.split(" "))
    val unit: RDD[String] = rdd4.flatMap(_.split(" "))

//    rdd4.map(_.split(' ')).collect.foreach(println)
//    rdd4.flatMap(_.split(' ')).collect.foreach(println)

    val rdd5 = sc.parallelize(List(List("a b c", "a b b"),List("e f g", "a f g"), List("h i j", "a a b")))
    //flatMap后得到的是三个List(),对每个list取前N
    val value1: RDD[String] = rdd5.flatMap(_.take(3))
//    value1.collect().foreach(println)
    val unit1: RDD[String] = rdd5.flatMap(_.flatMap(_.split(" ").take(1)))
//    unit1.collect().foreach(println)

    //union求并集，注意类型要一致
    val rdd6 = sc.parallelize(List(5,6,4,7))
    val rdd7 = sc.parallelize(List(1,2,3,4))
    val rdd8 = rdd6.union(rdd7)
//    rdd8.distinct.sortBy(x=>x).collect.foreach(print)
//    rdd8.collect().foreach(print)
//    rdd8.distinct().sortBy(x=>x,false).collect().foreach(print)

    // 求交集
    val interRDD: RDD[Int] = rdd6.intersection(rdd7)
//    interRDD.collect().foreach(print)

    //join  (tom,(1,8))(tom,(1,2))(jerry,(2,9))
    val rdd1 = sc.parallelize(List(("tom", 1), ("jerry", 2), ("kitty", 3)))
    val rdd2 = sc.parallelize(List(("jerry", 9), ("tom", 8), ("shuke", 7), ("tom", 2)))
    val joinRDD: RDD[(String, (Int, Int))] = rdd1.join(rdd2)
    //    joinRDD.collect().foreach(print)
    //groupByKey() (tom,CompactBuffer((1,8), (1,2)))(jerry,CompactBuffer((2,9)))
    val groupByKey: RDD[(String, Iterable[(Int, Int)])] = joinRDD.groupByKey()

    //    groupByKey.collect().foreach(print)
//    val value2: RDD[(String, (Int, Int))] = groupByKey.map(x => (x._1, x._2.sum))
//    groupByKey.mapValues(_.sum).collect().foreach(print)
//    value2.collect().foreach(print)
    //左外join
    val leftOuterJoin: RDD[(String, (Int, Option[Int]))] = rdd1.leftOuterJoin(rdd2)

    //右外join
    val rightOuterJoin: RDD[(String, (Option[Int], Int))] = rdd1.rightOuterJoin(rdd2)

    //reduce
    val rdd10 = sc.parallelize(List(1,2,3,4,5), 2)
    val i: Int = rdd10.reduce(_ + _)

    println(i)
  }
}
